package com.my.study.config;

import com.my.study.common.constant.TopicConstants;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.CreateTopicsResult;
import org.apache.kafka.clients.admin.ListTopicsResult;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.internals.Topic;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ExecutionException;

/**
 * kafka初始化topic
 *
 * https://blog.csdn.net/Davis_Dxs/article/details/88866569
 *
 * @author: yidujun
 * @create: 2021/07/21 16:10
 */
@Slf4j
@Configuration
public class KafkaInitialTopicConfiguration {


    @Value("${kafka.topic}")
    private String topics;

    @Resource
    AdminClient adminClient;


    @PostConstruct
    public void initTopic() throws Exception {
        String[] splitTopics = topics.split(",");
        ListTopicsResult listTopicsResult = adminClient.listTopics();
        KafkaFuture<Set<String>> names = listTopicsResult.names();
        Set<String> strings = names.get();

        List<NewTopic> list = new ArrayList<>();
        for (String splitTopic : splitTopics) {
            if (!strings.contains(splitTopic)) {
                NewTopic newTopic = new NewTopic(splitTopic, 9, (short) 1);
                list.add(newTopic);
            }
        }

        CreateTopicsResult topics = adminClient.createTopics(list);

        try {
            topics.all().get();
        } catch (InterruptedException | ExecutionException e) {
            log.error("初始化主题错误", e);
            e.printStackTrace();
        }

    }

//    @Override
//    public void run(String... args) throws Exception {
//        String[] splitTopics = topics.split(",");
//        ListTopicsResult listTopicsResult = adminClient.listTopics();
//        KafkaFuture<Set<String>> names = listTopicsResult.names();
//        Set<String> strings = names.get();
//
//        List<NewTopic> list = new ArrayList<>();
//        for (String splitTopic : splitTopics) {
//            if (!strings.contains(splitTopic)) {
//                NewTopic newTopic = new NewTopic(splitTopic, 9, (short) 1);
//                list.add(newTopic);
//            }
//        }
//
//        CreateTopicsResult topics = adminClient.createTopics(list);
//
//        try {
//            topics.all().get();
//        } catch (InterruptedException | ExecutionException e) {
//            log.error("初始化主题错误", e);
//            e.printStackTrace();
//        }
//    }


    /**
     * 初始化一个名为topic01的主题，分区2个，副本2个
     *
     * @return
     */
    @Bean
    public NewTopic init() {
        return new NewTopic("topic10", 9, (short) 1);
    }

}
